package com.bonc.wkafka.test;

import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

public class ConsumerTest4 {
	public static void main(String[] args) {
		Properties p = new Properties();
		p.put(ConsumerConfig.GROUP_ID_CONFIG, "11sw03");
		p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop05:9092");
		p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
		KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(p, new StringDeserializer(), new StringDeserializer());
		consumer.subscribe(Collections.singleton("w2"), new ConsumerRebalanceListener() {
			
			@Override
			public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
				Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<TopicPartition, OffsetAndMetadata>();
				// TODO Auto-generated method stub
				System.out.println("Partitions Revoked");
				Iterator<TopicPartition> iterator = partitions.iterator();
				while(iterator.hasNext()) {
					TopicPartition tp = iterator.next();
					long offset = consumer.position(tp);
					System.out.println("tp: "+tp.topic()+"-"+tp.partition()+", offset: "+offset);
					offsets.put(tp, new OffsetAndMetadata(offset));
				}
				consumer.commitAsync(offsets, null);
			}
			
			@Override
			public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
				// TODO Auto-generated method stub
				System.out.println("Partitions Assigned");
				Iterator<TopicPartition> iterator = partitions.iterator();
				while(iterator.hasNext()) {
					TopicPartition tp = iterator.next();
					OffsetAndMetadata committed = consumer.committed(tp);
					System.out.println("tp: "+tp.topic()+"-"+tp.partition()+", offset: "+committed.offset());
				}
			}
		});
		while(true) {
			ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(3000));
			Iterator<ConsumerRecord<String, String>> iterator = records.iterator();
			while(iterator.hasNext()) {
				ConsumerRecord<String, String> record = iterator.next();
				System.out.println("partition: "+record.partition()+", value: "+record.value());
			}
		}
	}
}
